package akka.streams

import java.nio.file.Paths

import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._
import akka.util.ByteString
import akka.{Done, NotUsed}

import scala.concurrent.ExecutionContext.Implicits._
import scala.concurrent.Future
import scala.util.{Failure, Success}

/**
  * Created by shichen on 2017/6/23.
  * https://tech.zalando.com/blog/about-akka-streams/?gh_src=4n3gxh1
  */
object AkkaStreamTest extends App{
  //t2()
  //simpleSource()
  ch6()

  def simpleSource(): Unit = {
    val source: Source[Int,NotUsed] = Source(1 to 100)
    implicit val sys = ActorSystem("QuickStart")
    implicit val materializer = ActorMaterializer()

    //val done: Future[Done] = source.runForeach(i => println(i))(materializer)

    val factorials = source.scan(BigInt(1))((acc,next) => acc * next)
    /*
    val result: Future[IOResult] = factorials.map(num => ByteString(s"$num\n"))
      .runWith(FileIO.toPath(Paths.get("target/factorials.txt")))
    */
    val result: Future[IOResult] = factorials.map(_.toString).runWith(lineSink("target/factorial2.txt"))

    result.onComplete(_ => sys.terminate())


  }
  def lineSink(filename: String): Sink[String,Future[IOResult]] = Flow[String]
    .map(s => ByteString(s + "\r\n"))
    .toMat(FileIO.toPath(Paths.get(filename)))(Keep.right)

  def t2(): Unit = {

    implicit val sys = ActorSystem("akka-stream-example")
    implicit val materializer = ActorMaterializer()

    val helloWorldStream: RunnableGraph[Future[Done]] =
      Source.single("Hello world")
        .map(s => s.toUpperCase())
        .toMat(Sink.foreach(println))(Keep.right)
    val doneF: Future[Done] = helloWorldStream.run()
    doneF.onComplete {
      case Success(Done) =>
        println("Stream finished successfully.")
      case Failure(e) =>
        println(s"Stream failed with $e")
    }
  }

  def t1(): Unit = {
    //a single string value goes from its Source through a mapping stage
    // Flow[String].map and ends up in a Sink that printlns its input.
    val helloWorldStream: RunnableGraph[NotUsed] =
      Source.single("hello world")
        .via(Flow[String].map(s => s.toUpperCase()))
        .to(Sink.foreach(println))

    val helloWorldStream2: RunnableGraph[NotUsed] =
      Source.single("hello world2")
        .map(s => s.toUpperCase())
        .to(Sink.foreach(println))

    implicit val sys = ActorSystem("akka-stream-example")
    implicit val materializer = ActorMaterializer()
    helloWorldStream.run()

    val meterializedValue: NotUsed = helloWorldStream2.run()

    sys.terminate()
  }

  def t3(): Unit = {
    import GraphDSL.Implicits._
    RunnableGraph.fromGraph(GraphDSL.create() { implicit builder =>
      val A: Outlet[Int]                  = builder.add(Source.single(0)).out
      val B: UniformFanOutShape[Int, Int] = builder.add(Broadcast[Int](2))
      val C: UniformFanInShape[Int, Int]  = builder.add(Merge[Int](2))
      val D: FlowShape[Int, Int]          = builder.add(Flow[Int].map(_ + 1))
      val E: UniformFanOutShape[Int, Int] = builder.add(Balance[Int](2))
      val F: UniformFanInShape[Int, Int]  = builder.add(Merge[Int](2))
      val G: Inlet[Any]               	= builder.add(Sink.foreach(println)).in
      C 	<~      F
      A  ~>  B  ~>  C 	~>      F
      B  ~>  D  ~>  E  ~>  F
      E  ~>  G
      ClosedShape
    })
  }


  //from mastering akka
  def ch6(): Unit = {
    val source:Source[Int,NotUsed] = Source(1 to 5)
    val sink: Sink[Int,Future[Done]] = Sink.foreach(println)
    val dataFlow:RunnableGraph[NotUsed] = source.to(sink)

    //the above is just a blueprint
    implicit val system = ActorSystem()
    implicit val mater = ActorMaterializer()
    dataFlow.run()

    val sink2:Sink[Int,Future[Int]] = Sink.fold(0)(_ + _)
    val dataFlow2:RunnableGraph[Future[Int]] = source.toMat(sink2)(Keep.right)
    val fut:Future[Int] = dataFlow2.run()
    fut.onComplete(println)//(system.dispatcher)

    val flow = Flow[Int].map(_ * 2).filter(_ % 2 == 0)
    val fut2 = source.via(flow).toMat(sink2)(Keep.right).run()
    fut2.onComplete(println)//(system.dispatcher)

    val flow2 = Flow[Int].map(_ * 3).filter(_ % 2 == 0)
    val fused = Fusing.aggressive(flow2)
    Source(List(1,2,3,4,5)).via(fused).runForeach(println)

    //每一个element从source到sink，顺序地
    Source(1 to 5).map{ x => println(s"pre-map:$x"); x}
      .map(_*3)
      .map { x => println(s"pre-filter:$x"); x}
      .filter(_ % 2 == 0)
      //.runForeach(x => s"done:$x")

    println("async flow...")
    Source(1 to 5).map{ x => println(s"async pre-map:$x"); x}
      .map(_*3).async //async前是并行的还是后面是并行的？
      .map { x => println(s"async pre-filter:$x"); x}
      .filter(_ % 2 == 0)
      .runForeach(x => println(s"async done:$x"))


  }
}
